package com.zyx.pulsardemo.connector.source;

import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.io.core.PushSource;
import org.apache.pulsar.io.core.SourceContext;
import org.slf4j.Logger;


/**
 * @author Yaxi.Zhang
 * @since 2021/10/12 14:39
 */
public class DirectorySource extends PushSource<String> {


    /**
     * 用于运行后台线程的内部线程池
     */
    private final ScheduledExecutorService scheduler =
            Executors.newScheduledThreadPool(1);

    private DirectoryConsumerThread scanner;

    private Logger log;

    /**
     * 检索用户提供的配置属性指定的本地目录读取文件
     * 然后启动一个后台线程类型DirectoryConsumerThread负责扫描目录和逐行读取每个文件的内容
     */
    @Override
    public void open(Map<String, Object> config, SourceContext context) throws Exception {
        log = context.getLogger();
        String inputDir = (String) config.getOrDefault("inputDir", ".");
        // 从传入的配置属性 获取运行时设置
        String processedDir = (String) config.getOrDefault("processedDir", ".");
        String frequency = (String) config.getOrDefault("frequency", "10");
        // 创建后台线程, 将引用传递给源连接器和配置
        scanner = new DirectoryConsumerThread(this, inputDir, processedDir, log);
        // 定时执行任务
        scheduler.scheduleAtFixedRate(scanner, 0, Long.parseLong(frequency), TimeUnit.MINUTES);
        log.info(String.format("Scheduled to run every %s minutes", frequency));
    }

    @Override
    public void close() throws Exception {
        log.info("Closing connector");
        scheduler.shutdownNow();
    }

}
